package com.gandong8.es.tool.service.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.PropertyNamingStrategy;
import com.gandong8.es.tool.core.EsDoc;
import com.gandong8.es.tool.core.EsDocId;
import com.gandong8.es.tool.core.EsQuery;
import com.gandong8.es.tool.service.EsService;
import com.gandong8.es.tool.util.EsUtil;
import lombok.extern.slf4j.Slf4j;
//import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
//version 7.x
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;

import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;

import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.script.Script;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;

@Service
@Slf4j
public class EsServiceImpl implements EsService {
    @Resource(name = "restHighLevelClient")
    RestHighLevelClient client;
    @Autowired
    BulkProcessor bulkProcessor;
    @Resource(name = "esJackson")
    ObjectMapper objectMapper;

    /**
     * 根据指定Class创建索引
     * @param  clazz
     * @return boolean
     */
    @Override
    public boolean createIndex(Class<?> clazz) {
        try {
            CreateIndexRequest request = EsUtil.classToRequest(clazz);
            return createIndex(request);
        } catch (Exception e) {
            log.error("createIndex_err:{}", clazz, e);
            return false;
        }
    }

    /**
     * 根据CreateIndexRequest创建索引
     * @param request
     * @return boolean
     */
    @Override
    public boolean createIndex(CreateIndexRequest request) {
        if(request == null) {
            return false;
        }
        try {
            client.indices().create(request, RequestOptions.DEFAULT);
            return true;
        } catch (Exception e) {
            //ConnectException
            //ElasticsearchStatusException
            log.error("createIndex_err", e);
            return false;
        }
    }

    /**
     * 删除索引
     * @param clazz
     * @return boolean
     */
    @Override
    public boolean deleteIndex(Class<?> clazz) {
        return deleteIndex(getIndexName(clazz));
    }

    @Override
    public boolean deleteIndex(String indexName) {
        try {
            DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(indexName);
            AcknowledgedResponse response = client.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
            if(response.isAcknowledged()) {
                return true;
            }
            log.error("deleteIndex_failed:{}", indexName);
        } catch (Exception e) {
            log.error("deleteIndex_err:{}", indexName, e);
        }
        return false;
    }


    @Override
    public boolean addDoc(EsDoc obj) {
        try {
            //id为null时IndexRequest.process方法会自动生成id:UUIDs.base64UUID()
            IndexRequest request = new IndexRequest(getIndexName(obj)).id(obj.esDocumentId())
                    .source(objectMapper.writeValueAsBytes(obj), XContentType.JSON);
            IndexResponse response = client.index(request, RequestOptions.DEFAULT);
            if(RestStatus.CREATED.equals(response.status())) {
                obj.setDocumentId(response.getId());
                return true;
            }
            if(RestStatus.OK.equals(response.status())) {
                obj.setDocumentId(response.getId());
                log.warn("update doc:{}", obj);
                return true;
            }
            log.error("addDocErr:{},{}", obj, response.status());
        } catch (Exception e) {
            log.error("addDocException:{}", obj, e);
        }
        return false;
    }

    @Override
    public boolean addDoc(String indexName, String docId, Map<String, Object> map) {
        try {
            IndexRequest post = new IndexRequest(indexName).id(docId)
                    .source(map);
            IndexResponse response = client.index(post, RequestOptions.DEFAULT);
            if(RestStatus.CREATED.equals(response.status())) {
                return true;
            }
            if(RestStatus.OK.equals(response.status())) {
                log.warn("update doc:{}", docId);
                return true;
            }
            log.error("addDocErr2:"+ indexName + ";"+ docId + ";"+response.status() + ";" + response.getResult());
        } catch (Exception e) {
            log.error("addDoc_err2:{}", indexName, e);
        }
        return false;
    }

    @Override
    public void bulkAddDoc(EsDoc obj) throws JsonProcessingException {
        IndexRequest request = new IndexRequest(getIndexName(obj)).id(obj.esDocumentId())
                .source(objectMapper.writeValueAsBytes(obj), XContentType.JSON);
        bulkProcessor.add(request);
    }

    @SuppressWarnings("unchecked")
    @Override
    public   <T> T getDoc(EsDocId obj) {
        return (T) getDoc(obj.esDocumentId(), (Class<? extends EsDoc>) obj.getClass());
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T extends EsDoc> T getDoc(String docId, Class<? extends EsDoc> clazz) {
        try {
            GetRequest getRequest = new GetRequest(getIndexName(clazz), docId);
            //GetRequest getRequest = new GetRequest(getIndexName(clazz), getIndexType(clazz), docId);
            GetResponse response = client.get(getRequest, RequestOptions.DEFAULT);
            if(!response.isExists()) {
                return null;
            }
            T obj = (T) objectMapper.readValue(response.getSourceAsBytes(), clazz);
            if(obj.esDocumentId() == null) {
                obj.setDocumentId(response.getId());
            }
            return  obj;
        } catch (Exception e) {
            log.error("getDoc_err:{}", docId, e);
        }
        return null;
    }

    @Override
    public boolean delDoc(EsDoc obj) {
        if(obj.esDocumentId() == null) {
            return false;
        }
        try {
            DeleteRequest request = new DeleteRequest(getIndexName(obj), obj.esDocumentId());
            //DeleteRequest request = new DeleteRequest(getIndexName(obj), getIndexType(obj), obj.esDocumentId());
            DeleteResponse response = client.delete(request, RequestOptions.DEFAULT);
            if(RestStatus.OK.equals(response.status())) {
                return true;
            }
            if(RestStatus.NOT_FOUND.equals(response.status())) {
                log.warn("doc not found:{}", obj);
                return false;
            }
            log.error(response.status() + ";" + response.getResult());
        } catch (Exception e) {
            log.error("delDoc:{}", obj, e);
        }
        return false;
    }

    @Override
    public void bulkDelDoc(EsDoc obj) {
        bulkProcessor.add(new DeleteRequest(getIndexName(obj), obj.esDocumentId()));
        //bulkProcessor.add(new DeleteRequest(getIndexName(obj), getIndexType(obj), obj.esDocumentId()));
    }

    @Override
    public boolean updateDoc(EsDoc obj) {
        try {
            UpdateRequest updateRequest = new UpdateRequest(getIndexName(obj), obj.esDocumentId());
            //UpdateRequest updateRequest = new UpdateRequest(getIndexName(obj), getIndexType(obj), obj.esDocumentId());
            updateRequest.doc(objectMapper.writeValueAsBytes(obj), XContentType.JSON);
            UpdateResponse response = client.update(updateRequest, RequestOptions.DEFAULT);
            if(RestStatus.OK.equals(response.status())) {
                return true;
            }
            log.error(response.status() + ";" + response.getResult());
        } catch (Exception e) {
            log.error("updateDoc:{}", obj, e);
        }
        return false;
    }



    public boolean updateDoc(String indexName, String docId, Map<String, Object> map) {
        try {
            UpdateRequest updateRequest = new UpdateRequest(indexName,  docId);
            updateRequest.doc(map);
            UpdateResponse response = client.update(updateRequest, RequestOptions.DEFAULT);
            if(RestStatus.OK.equals(response.status())) {
                return true;
            }
            log.error(response.status() + ";" + response.getResult());
        } catch (Exception e) {
            log.error("updateDoc:{}", docId, e);
        }
        return false;
    }

    public boolean updateByQuery(String indexName, Map<String, Object> queryMap, Map<String, Object> updateMap) throws IOException {
        BoolQueryBuilder queryBuilder = QueryBuilders.boolQuery();
        queryMap.forEach((key, value) -> queryBuilder.must(QueryBuilders.termQuery(key, value)));
        StringBuilder str = new StringBuilder();
        Set<String> keys = updateMap.keySet();
        for (String key : keys) {
            Object value = updateMap.get(key);
            String appendValue;
            if (value instanceof Integer) {
                appendValue = value.toString();
            } else if (value instanceof Long) {
                appendValue = value.toString();
            } else {
                appendValue = value.toString();
            }
            str.append("ctx._source.").append(key).append("=").append(appendValue).append(";");
        }
        UpdateByQueryRequest request = new UpdateByQueryRequest(indexName);
        request.setQuery(queryBuilder);
        request.setScript(new Script(str.toString()));
        client.updateByQuery(request, RequestOptions.DEFAULT);
        return true;
    }




    @Override
    public void bulkUpdateDoc(EsDoc obj) {
        bulkProcessor.add(new UpdateRequest(getIndexName(obj), obj.esDocumentId()));
        //bulkProcessor.add(new UpdateRequest(getIndexName(obj), getIndexType(obj),obj.esDocumentId()));
    }

    @SuppressWarnings("unchecked")
    @Override
    public <T extends EsDoc> List<T> search(EsQuery<T> query) throws IOException {
        SearchRequest searchRequest = query.search();
        //6.X.X to 7.X.X: contains unrecognized parameter: [ccs_minimize_roundtrips]]
        //在数据index后有refresh_interval这个间隔内search是不可见的,可以用get by id,或refresh=wait_for
        //https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-refresh.html
        SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = searchResponse.getHits();
        long totalHits = hits.getTotalHits().value;
        //long totalHits = hits.getTotalHits();
        query.setTotalHits(totalHits);
        float maxScore = hits.getMaxScore();
        SearchHit[] searchHits = hits.getHits();
        List<T> objList = new ArrayList<>();
        objectMapper.setPropertyNamingStrategy(PropertyNamingStrategy.LOWER_CAMEL_CASE);
        for (SearchHit hit : searchHits) {
            String str = hit.getSourceAsString();
            T obj = (T) objectMapper.readValue(hit.getSourceAsString(), query.getObj().getClass());
            obj.setDocumentId(hit.getId());
            objList.add(obj);
        }
        return objList;
    }


    private String getIndexType(Class<?> clazz) {
        //todo cache
        return EsUtil.parseIndexType(clazz);
    }

    private String getIndexType(EsDoc obj) {
        return getIndexType(obj.getClass());
    }

    private String getIndexName(Class<?> clazz) {
        //todo cache
        return EsUtil.parseIndexName(clazz);
    }

    private String getIndexName(EsDoc obj) {
        return getIndexName(obj.getClass());
    }



}
